# -*- coding: utf-8 -*-

"""
calculate the diff of two redis json file
"""

from __future__ import unicode_literals
from __future__ import absolute_import
from __future__ import division

from kgpipeline.job import KgJob, JobInputMisMatchError
from kgpipeline.jsonutil import entity_json_decode, entity_json_encode
from kgpipeline.sparkutil import to_text_dataframe

from typing import List, Tuple, Dict
from pyspark.sql import DataFrame


def is_list_same(list1, list2):
    dict1 = {}
    dict2 = {}
    for e in list1:
        dict1[e["@id"]] = e
    for e in list2:
        dict2[e["@id"]] = e

    if set(dict1.keys()) != set(dict2.keys()):
        return False
    else:
        for key, v1 in dict1.items():
            if v1 != dict2[key]:
                return False
        return True


def diff_one(x):
    name = x[0]
    list1 = x[1][0]
    list2 = x[1][1]
    if list1 is None:
        return [(name, None)]
    elif list2 is None:
        return [(name, list1)]
    else:
        if not is_list_same(list1, list2):
            return [(name, None), (name, list1)]
        else:
            return []


class KgRedisJsonDiffJob(KgJob):
    def process(self, inputs):  # type: (KgRedisJsonDiffJob, List[Tuple[DataFrame,Dict]]) -> (List[DataFrame])
        if len(inputs) != 2:
            raise JobInputMisMatchError("needs exactly 2 input json files for a Redis Json Diff Job")

        name1, name2 = [df.rdd.map(lambda x: entity_json_decode(x["value"])) for df, _ in inputs]

        diff = name1.fullOuterJoin(name2).flatMap(diff_one)

        return [to_text_dataframe(diff, entity_json_encode)]


KgJob.register("redisjsondiff", KgRedisJsonDiffJob)
